package com.exsun.gateway.jt809.master.handler;

import com.exsun.gateway.jt809.common.LinkType;
import com.exsun.gateway.jt809.kafka.KafkaProducer;
import com.exsun.gateway.jt809.data.PackageData;
import com.exsun.gateway.jt809.common.LinkMgr;
import com.exsun.gateway.jt809.config.NettyClientConfig;
import com.exsun.gateway.jt809.handler.ResponseHandler;
import com.exsun.gateway.jt809.handler.SendMgrHandler;
import com.exsun.gateway.jt809.master.JT809Client;
import exsun.jt808.data.jt809_data.JT809LinkStatus;
import exsun.jt808.data.util.executorsThread.ScheduledHandle;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

@Component
@Sharable
public class JT809ClientHandler extends SimpleChannelInboundHandler<PackageData> {

	private final static Logger logger = Logger.getLogger(JT809ClientHandler.class);

	@Autowired
	ResponseHandler responseHandler;
	@Autowired
	NettyClientConfig nettyClientConfig;
	@Autowired
	private SendMgrHandler sendMgrHandler;

	private JT809Client jt809Client;

	@Autowired
	KafkaProducer kafkaProducer;

	public void setJT809Client(JT809Client client){
		this.jt809Client = client;
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, PackageData pkg) throws Exception {
		try {
			logger.info("收到来自上级平台的消息：" + pkg.getMsgBodyBytes());
			PackageData pak = responseHandler.response(pkg);
			if(pak == null)return;
			LinkMgr.push2FitChannel(pak.getMsgHeader().getMsgId(), pak.getMsgBodyBytes(), pak.getLinkType());

		} finally {
			release(pkg);
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception{
//		ctx.close();
		logger.error("异常：{}", cause);
	}

	/**
	 * 主链路登录请求消息
	 *
	 * @param ctx
	 * @return
	 */
	@Override
	public void channelActive(final ChannelHandlerContext ctx) throws Exception {
		LinkMgr.setMasterChannel(ctx.channel());
		//主链路登录请求
		masterLoggingRequest();
		//将链路状态推送到kafka
		kafkaStatus(JT809LinkStatus.MasterOnline);
	}

	static ScheduledFuture<?> masterLoggingRequestScheduledFuture = null;
	/**
	 * 若登录失败定时发送主链路登录请求
	 */
	public synchronized void masterLoggingRequest(){
		if(masterLoggingRequestScheduledFuture != null && !masterLoggingRequestScheduledFuture.isCancelled()){
			return;
		}
		Runnable runnable = new Runnable() {
			public void run() {
				try {
					if(masterLoggingRequestScheduledFuture != null && LinkMgr.isMasterChannelLoggingStatus()){
						masterLoggingRequestScheduledFuture.cancel(true);
						masterLoggingRequestScheduledFuture = null;
						return;
					}
					//主链路登录请求
					sendMgrHandler.upConnectReq(nettyClientConfig);
				}catch (Exception e){
					logger.error(e.getMessage());
				}
			}
		};
		// 第二个参数为首次执行的延时时间，第三个参数为定时执行的间隔时间
		masterLoggingRequestScheduledFuture = ScheduledHandle.getInst().getService().scheduleAtFixedRate(runnable, 0, 5, TimeUnit.SECONDS);

	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) {
		try {
			logger.info("断开连接");
			LinkMgr.setMasterChannel(null);
			LinkMgr.setMasterChannelLoggingStatus(false);
			//将链路状态推送到kafka
			kafkaStatus(JT809LinkStatus.MasterOffline);
			//重连操作
			jt809Client.connect();

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (evt instanceof IdleStateEvent) {
			IdleState state = ((IdleStateEvent) evt).state();
			if (state.equals(IdleState.WRITER_IDLE)) {
				logger.error("主链路写数据超时");
				ctx.close();
			}else if(state.equals(IdleState.ALL_IDLE)) {
				//在没有数据往来的 发送主链路连接保持请求消息
				sendMgrHandler.linkHeart(ctx.channel());
			}
		}
	}

	private void release(Object msg) {
		try {
			ReferenceCountUtil.release(msg);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void kafkaStatus(JT809LinkStatus status){
		kafkaProducer.kafkaStatus(status);
	}

}
